package com.study.flink.demo;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import com.study.flink.entity.does.TicketsPricingDO;
import com.study.flink.source.TicketsPricingSource;

import lombok.extern.slf4j.Slf4j;

/**
 * 通过TicketsPricingSource获取数据文件中的数据，进行简单操作。
 * TicketsPricingSource只是简单的通过“,”作为分隔符识别每项数据。对应的是TicketsPricingCsvDemo中以CSV格式读取数据。
 * 
 * @projectName: flink-training-exercises
 * @author: yandong
 * @description:
 * @date: 2019年7月15日 下午2:54:00
 * @version:
 */
@Slf4j
public class TicketsPricingDemo {
    public void run() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        try {
            env.addSource(new TicketsPricingSource())
                .keyBy("trainType").map(new TicketsPricingRichMapFunction())
                .writeAsText("/data/bigdata/forflink/ticketspricing."
                + new SimpleDateFormat("YYMMddHHmmss").format(new Date()));
            ;

            env.execute();
        }
        catch (Exception e) {
            e.printStackTrace();
        }

        log.info("已经结束");
    }

    @SuppressWarnings("serial")
    public static class TicketsPricingRichMapFunction
        extends RichMapFunction<TicketsPricingDO, Tuple2<String, Integer>> {
        private ValueState<Integer> valueState = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            ValueStateDescriptor<Integer> descriptor = new ValueStateDescriptor<>("moving count",
                Integer.class);
            valueState = getRuntimeContext().getState(descriptor);
            super.open(parameters);
        }

        @Override
        public Tuple2<String, Integer> map(TicketsPricingDO value) throws Exception {

            valueState.update(null == valueState.value() ? 1 : valueState.value() + 1);
            log.info(String.format("[%s, %s, %d]", value.getId(), value.getTrainType(),
                valueState.value()));
            return new Tuple2<String, Integer>(value.getTrainType(), valueState.value());
        }
    }
}
